package org.huangrui.spark.java.sql;

import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;

/**
 * @Author hr
 * @Create 2024-10-21 7:00
 */
public class SparkSQL09_Source_Req_2 {
    public static void main(String[] args) {
        // TODO 在编码前，设定Hadoop的访问用户
        System.setProperty("HADOOP_USER_NAME", "huangrui");
        // TODO 启用Hive的支持
        SparkSession spark = SparkSession.builder().enableHiveSupport().appName("SparkSQL08_Source_Hive").master("local[2]").getOrCreate();
        spark.sql("use db_spark");
        spark.udf().register("cityRemark", functions.udaf(new MyCityRemarkUDAF(), Encoders.STRING()));
        spark.sql("SELECT\tc.area,\tp.product_name,\tcount(*) clickCnt,cityRemark(city_name) cityremark\n"
                + "FROM\n"
                + "\t( SELECT click_product_id, city_id FROM user_visit_action WHERE click_product_id != - 1 ) AS a\n"
                + "\tJOIN product_info p ON a.click_product_id = p.product_id\n"
                + "\tJOIN ( SELECT city_id, city_name, area FROM city_info ) c ON a.city_id = c.city_id \n"
                + "GROUP BY\tarea,\tproduct_id,\tproduct_name \n"
                + "LIMIT 10").show();

        spark.sql("select\n"
                + "     *\n"
                + " from (\n"
                + "     select\n"
                + "         *,\n"
                + "         rank() over( partition by area order by clickCnt desc ) as rank\n"
                + "     from (\n"
                + "         select\n"
                + "            area,\n"
                + "            product_name,\n"
                + "            count(*) as clickCnt,\n"
                + "            cityRemark(city_name) cityremark\n"
                + "         from (\n"
                + "             select\n"
                + "                a.*,\n"
                + "                p.product_name,\n"
                + "                c.area,\n"
                + "                c.city_name\n"
                + "             from user_visit_action a\n"
                + "             join product_info p on a.click_product_id = p.product_id\n"
                + "             join city_info c on a.city_id = c.city_id\n"
                + "             where a.click_product_id > -1\n"
                + "         ) t1 group by area, product_name\n"
                + "     ) t2\n"
                + " ) t3 where rank <= 3").show(false);

        spark.stop();
    }
}
